package com.es.service;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.es.exception.EsException;
import com.es.pojo.EsEntity;
import com.es.pojo.ReindexReq;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.ReindexRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @ClassName EsService
 * @author: maliang
 * @Description 逻辑层
 * @date 2021/11/26 17:18
 * @Version 1.0版本
 */
@Service
public class EsService {

    private static final String INDEX_EXCEP = "index is not exists";

    private static final String ID_EXCEP = "id is not exists";

    private static final String FROM_INDEX_EXCEP = "from index is not exists";

    private static final String TO_INDEX_EXCEP = "to index is not exists";

    @Resource
    private RestHighLevelClient restHighLevelClient;

    /**
     * 创建索引
     *
     * @param esEntity 搜索信息
     *                 String index
     *                 List<Property> field
     * @return 是否成功
     */
    public boolean createIndex(EsEntity esEntity) {
        String index = esEntity.getIndex();
        List<EsEntity.Property> fields = esEntity.getField();
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (CollectionUtils.isEmpty(fields)) {
            throw new EsException("fields is not exists");
        }
        Map<String, Map<String, String>> properties = new HashMap<>();
        for (EsEntity.Property field : fields) {
            Map<String, String> fieldMap = Collections.singletonMap("type", field.getType());
            properties.put(field.getName(), fieldMap);
        }
        Map<String, Map<String, Map<String, String>>> mapping = new HashMap<>();
        mapping.put("properties", properties);
        CreateIndexRequest request = new CreateIndexRequest(index);
        //2.客户端执行请求
        try {
            request.mapping(mapping);
            CreateIndexResponse createIndexResponse = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
            return createIndexResponse.isAcknowledged();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 删除索引
     *
     * @param index 索引名称
     * @return 是否成功
     */
    public boolean deleteIndex(String index) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        DeleteIndexRequest deleteRequest = new DeleteIndexRequest(index);
        //以TimeValue形式设置超时
        deleteRequest.timeout(TimeValue.timeValueMinutes(2));
        try {
            AcknowledgedResponse delete = restHighLevelClient.indices().delete(deleteRequest, RequestOptions.DEFAULT);
            return delete.isAcknowledged();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 根据id删除数据
     *
     * @param index 索引名称
     * @param id    数据id
     * @return 是否成功
     */
    public String deleteData(String index, String id) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        DeleteRequest deleteRequest = new DeleteRequest(index);
        deleteRequest.id(id);
        //以TimeValue形式设置超时
        deleteRequest.timeout(TimeValue.timeValueMinutes(2));
        //以WriteRequest.RefreshPolicy实例的形式设置刷新策略
        deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        try {
            DeleteResponse delete = restHighLevelClient.delete(deleteRequest, RequestOptions.DEFAULT);
            return delete.getResult().getLowercase();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "fail";
    }

    /**
     * 分析索引是否存在
     *
     * @param index 索引名称
     * @return 是否成功
     */
    public boolean exists(String index) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        GetIndexRequest getRequest = new GetIndexRequest(index);
        try {
            return restHighLevelClient.indices().exists(getRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 分析数据是否存在
     *
     * @param index 索引名称
     * @param id    数据id
     * @return 是否成功
     */
    public boolean exists(String index, String id) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        GetRequest getRequest = new GetRequest(index, id);
        try {
            return restHighLevelClient.exists(getRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 添加数据
     *
     * @param esEntity String index 索引名称
     *                 String entity json字符串
     * @return 是否成功
     */
    public String addData(EsEntity esEntity) {
        String index = esEntity.getIndex();
        String entity = esEntity.getEntity();
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(entity)) {
            throw new EsException("entity is not exists");
        }
        IndexRequest indexRequest = new IndexRequest(index);
        Map<String, String> map = JSONUtil.toBean(entity, Map.class);
        String id = map.get("id");
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        indexRequest.id(id);
        indexRequest.source(entity, XContentType.JSON);
        //客户端发送请求，获取响应结果
        try {
            IndexResponse indexResponse = restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
            return indexResponse.getResult().getLowercase();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "fail";
    }

    /**
     * 根据id查找数据
     *
     * @param index 索引名称
     * @param id    数据id
     * @return 数据的json字符串
     */
    public String getStrById(String index, String id) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        GetRequest getRequest = new GetRequest(index, id);
        try {
            GetResponse getResponsee = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            return getResponsee.getSourceAsString();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "not found";
    }

    /**
     * 根据id查找数据
     *
     * @param index 索引名称
     * @param id    数据id
     * @return 数据的json结构数据
     */
    public Map<String, Object> getMapById(String index, String id) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        GetRequest getRequest = new GetRequest(index, id);
        try {
            GetResponse getResponsee = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
            return getResponsee.getSourceAsMap();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Collections.emptyMap();
    }

    /**
     * 查找索引全部数据
     *
     * @param index 索引名称
     * @param from  起始数据索引
     * @param size  显示条数
     * @return 数据列表
     */
    public List<String> findAll(String index, Integer from, Integer size) {
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        SearchRequest searchRequest = new SearchRequest(index);
        try {
            SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
            searchSourceBuilder.from(from);
            searchSourceBuilder.size(size);
            MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();
            searchSourceBuilder.query(matchAllQueryBuilder);
            searchRequest.source(searchSourceBuilder);
            SearchResponse getResponsee = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHit[] hits = getResponsee.getHits().getHits();
            List<SearchHit> searchHits = Arrays.asList(hits);
            return searchHits.stream().map(SearchHit::getSourceAsString).collect(Collectors.toList());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }

    /**
     * 根据id更新数据
     *
     * @param esEntity 更新数据
     *                 String index 索引名称
     *                 String id 数据id
     *                 String entity 更新后的数据
     * @return 是否成功
     */
    public int updateById(EsEntity esEntity) {
        String index = esEntity.getIndex();
        String id = esEntity.getId();
        String entity = esEntity.getEntity();
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        if (StrUtil.isBlank(id)) {
            throw new EsException(ID_EXCEP);
        }
        if (StrUtil.isBlank(entity)) {
            throw new EsException("entity is not exists");
        }
        UpdateRequest updateRequest = new UpdateRequest(index, id);
        try {
            updateRequest.doc(entity, XContentType.JSON);
            UpdateResponse update = restHighLevelClient.update(updateRequest, RequestOptions.DEFAULT);
            return update.status().getStatus();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return 0;
    }

    /**
     * 根据条件查找数据(单一条件)
     *
     * @param esEntity 搜索条件
     *                 String index 索引名称
     *                 Map<String, String> criteria 搜索条件
     * @return 数据列表
     */
    public List<String> searchBySingleCondition(EsEntity esEntity) {
        String index = esEntity.getIndex();
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        SearchRequest searchRequest = new SearchRequest(index);
        Map<String, String> criteria = esEntity.getCriteria();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        if (!CollectionUtils.isEmpty(criteria)) {
            for (Map.Entry<String, String> entry : criteria.entrySet()) {
                MatchQueryBuilder termQueryBuilder = QueryBuilders.matchQuery(entry.getKey(), entry.getValue());
                searchSourceBuilder.query(termQueryBuilder);
            }
            searchSourceBuilder.from(esEntity.getFrom() == null ? 0 : esEntity.getFrom());
            searchSourceBuilder.size(esEntity.getSize() == null ? 10 : esEntity.getSize());
            searchSourceBuilder.sort("id", SortOrder.ASC);
        }
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse getResponsee = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHit[] hits = getResponsee.getHits().getHits();
            List<SearchHit> searchHits = Arrays.asList(hits);
            return searchHits.stream().map(SearchHit::getSourceAsString).collect(Collectors.toList());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }

    /**
     * 根据条件查找数据（多条件）
     *
     * @param esEntity 搜索条件
     *                 String index 索引名称
     *                 Map<String, String> criteria 搜索条件
     * @return 数据列表
     */
    public List<String> searchByCriteria(EsEntity esEntity) {
        String index = esEntity.getIndex();
        if (StrUtil.isBlank(index)) {
            throw new EsException(INDEX_EXCEP);
        }
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        Map<String, String> criteria = esEntity.getCriteria();
        if (!CollectionUtils.isEmpty(criteria)) {
            for (Map.Entry<String, String> entry : criteria.entrySet()) {
                MatchQueryBuilder matchQueryBuilder = QueryBuilders.matchQuery(entry.getKey(), entry.getValue());
                boolQueryBuilder.should(matchQueryBuilder);
            }
        }
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);
        searchSourceBuilder.from(esEntity.getFrom() == null ? 0 : esEntity.getFrom());
        searchSourceBuilder.size(esEntity.getSize() == null ? 10 : esEntity.getSize());
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse getResponsee = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            SearchHit[] hits = getResponsee.getHits().getHits();
            List<SearchHit> searchHits = Arrays.asList(hits);
            return searchHits.stream().map(SearchHit::getSourceAsString).collect(Collectors.toList());
        } catch (IOException e) {
            e.printStackTrace();
        }
        return Collections.emptyList();
    }

    /**
     * 批量插入数据
     *
     * @param esEntity 批量插入的数据
     *                 String index 索引名称
     *                 List<String> entitys 插入数据的json字符串
     * @return 是否成功
     */
    public boolean insertList(EsEntity esEntity) {
        String index = esEntity.getIndex();
        List<String> entitys = esEntity.getEntitys();
        BulkRequest bulkRequest = new BulkRequest();
        for (String entity : entitys) {
            Map<String, String> map = JSONUtil.toBean(entity, Map.class);
            String id = map.get("id");
            if (StrUtil.isNotBlank(id)) {
                IndexRequest source = new IndexRequest(index).id(id).source(entity, XContentType.JSON);
                bulkRequest.add(source);
            } else {
                IndexRequest source = new IndexRequest(index).source(entity, XContentType.JSON);
                bulkRequest.add(source);
            }
        }
        try {
            BulkResponse bulk = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
            return bulk.hasFailures();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * copy数据从指定索引到目标索引
     *
     * @param reindexReq 数据索引来源 数据索引去向
     * @return 是否成功
     */
    public String transferData(ReindexReq reindexReq) {
        if (CollectionUtils.isEmpty(reindexReq.getFrom())) {
            throw new EsException(FROM_INDEX_EXCEP);
        }
        if (StringUtils.isEmpty(reindexReq.getTo())) {
            throw new EsException(TO_INDEX_EXCEP);
        }
        ReindexRequest request = new ReindexRequest();
        String[] froms = reindexReq.getFrom().toArray(new String[0]);
        request.setSourceIndices(froms);
        request.setDestIndex(reindexReq.getTo());
        //设置versionType为EXTERNAL
        request.setDestVersionType(VersionType.EXTERNAL);
        //设置versionType为create
        request.setDestOpType("create");
        // 默认情况下_reindex单次可以处理1000个文档。您可以使用sourceBatchSize更改批处理大小。
        if (reindexReq.getSize() != null) {
            request.setSourceBatchSize(reindexReq.getSize());
        }
        //设置版本冲突时继续
        request.setConflicts("proceed");
        try {
            BulkByScrollResponse reindex = restHighLevelClient.reindex(request, RequestOptions.DEFAULT);
            return reindex.getStatus().getTotal() > 0 ? "suceess" : "fail";
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "fail";
    }

    /**
     * 复制数据从指定索引到目标索引并且删除原索引
     *
     * @param reindexReq 数据索引来源 数据索引去向
     * @return 是否成功
     */
    public String transferDataAndDelFrom(ReindexReq reindexReq) {
        if (CollectionUtils.isEmpty(reindexReq.getFrom())) {
            throw new EsException(FROM_INDEX_EXCEP);
        }
        if (StringUtils.isEmpty(reindexReq.getTo())) {
            throw new EsException(TO_INDEX_EXCEP);
        }
        ReindexRequest request = new ReindexRequest();
        String[] froms = reindexReq.getFrom().toArray(new String[0]);
        request.setSourceIndices(froms);
        request.setDestIndex(reindexReq.getTo());
        //设置versionType为EXTERNAL
        request.setDestVersionType(VersionType.EXTERNAL);
        //设置versionType为create
        request.setDestOpType("create");
        // 默认情况下_reindex单次可以处理1000个文档。您可以使用sourceBatchSize更改批处理大小。
        if (reindexReq.getSize() != null) {
            request.setSourceBatchSize(reindexReq.getSize());
        }
        //设置版本冲突时继续
        request.setConflicts("proceed");
        try {
            BulkByScrollResponse reindex = restHighLevelClient.reindex(request, RequestOptions.DEFAULT);
            for (String str : froms) {
                deleteIndex(str);
            }
            return reindex.getStatus().getTotal() > 0 ? "suceess" : "fail";
        } catch (IOException e) {
            e.printStackTrace();
        }
        return "fail";
    }
}